package cn.cuilan.common.core;

import cn.cuilan.common.entity.ProtocolMessage;
import cn.cuilan.common.pipeline.CompositeMessageHandler;
import io.netty.bootstrap.AbstractBootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.SocketAddress;
import java.util.concurrent.atomic.AtomicReference;

/**
 * 抽象协议连接器
 *
 * @param <C> 连接器配置
 * @param <B> Bootstrap启动类
 * @param <M> 报文消息实体对象
 * @author zhang.yan
 */
public abstract class AbstractProtocolConnector<
        C extends ConnectorConfiguration,
        B extends AbstractBootstrap<B, ?>,
        M extends ProtocolMessage> {

    protected final Logger logger = LoggerFactory.getLogger(getClass());

    /**
     * 自定义综合消息处理器
     */
    private final CompositeMessageHandler<M> messageHandler;

    /**
     * 确保Channel的原子操作
     */
    private final AtomicReference<Channel> channelRef = new AtomicReference<>();

    /**
     * 连接配置类
     */
    private final C configuration;

    private ConnectorConfigurer<C, B> configurer;
    private SocketAddress socketAddress;
    private EventLoopGroup bossEventLoopGroup;
    private EventLoopGroup workerEventLoopGroup;
    private B bootstrap;

    public AbstractProtocolConnector(C configuration) {
        this(configuration, new CompositeMessageHandler<>());
    }

    public AbstractProtocolConnector(C configuration, CompositeMessageHandler<M> messageHandler) {
        assert (configuration != null) : "Configuration must be provided";
        this.configuration = configuration;
        this.messageHandler = messageHandler;
    }

    /**
     * Making connector ready to create a connection / bind to port.
     * Creates a Bootstrap
     *
     * @see AbstractBootstrap
     */
    public void init() {
        logger.info("Initializing");
        bossEventLoopGroup = createBossEventLoopGroup();
        workerEventLoopGroup = createWorkerEventLoopGroup();
        bootstrap = createBootstrap();
    }

    /**
     * Shutdown connector if groups is not null.
     */
    public void shutdown() {
        if (workerEventLoopGroup != null) {
            workerEventLoopGroup.shutdownGracefully();
            workerEventLoopGroup = null;
        }
        if (bossEventLoopGroup != null) {
            bossEventLoopGroup.shutdownGracefully();
            bossEventLoopGroup = null;
        }
    }

    /**
     * 配置Bootstrap参数
     *
     * @param bootstrap ServerBootstrap or Bootstrap.
     */
    protected void configureBootstrap(B bootstrap) {
        bootstrap.option(ChannelOption.TCP_NODELAY,
                Boolean.parseBoolean(System.getProperty("nfs.rpc.tcp.nodelay", "true")))
                .option(ChannelOption.AUTO_READ, true);
        if (configurer != null) {
            configurer.configureBootstrap(bootstrap, configuration);
        }
    }

    protected EventLoopGroup createBossEventLoopGroup() {
        return new NioEventLoopGroup();
    }

    protected EventLoopGroup getBossEventLoopGroup() {
        return bossEventLoopGroup;
    }

    protected EventLoopGroup createWorkerEventLoopGroup() {
        // 设置工作线程数
        final NioEventLoopGroup group = new NioEventLoopGroup(configuration.getWorkerThreadsCount());
        logger.debug("Created worker EventLoopGroup with {} executor threads", group.executorCount());
        return group;
    }

    protected EventLoopGroup getWorkerEventLoopGroup() {
        return workerEventLoopGroup;
    }

    /**
     * 创建Bootstrap，子类实现
     */
    protected abstract B createBootstrap();

    public B getBootstrap() {
        return bootstrap;
    }

    public ConnectorConfigurer<C, B> getConfigurer() {
        return configurer;
    }

    public void setConfigurer(ConnectorConfigurer<C, B> configurer) {
        this.configurer = configurer;
    }

    public SocketAddress getSocketAddress() {
        return socketAddress;
    }

    public void setSocketAddress(SocketAddress socketAddress) {
        this.socketAddress = socketAddress;
    }

    public Channel getChannel() {
        return channelRef.get();
    }

    public void setChannel(Channel channel) {
        this.channelRef.set(channel);
    }

    public C getConfiguration() {
        return configuration;
    }

    public CompositeMessageHandler<M> getMessageHandler() {
        return messageHandler;
    }

    /**
     * 添加监听处理器
     */
    public void addMessageListener(ProtocolMessageListener<M> handler) {
        messageHandler.addListener(handler);
    }

    /**
     * 移除监听处理器
     */
    public void removeMessageListener(ProtocolMessageListener<M> handler) {
        messageHandler.removeListener(handler);
    }
}
